package com.learn.sink;

import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * @create: 2023-04-18 16:17
 * @author: Mr.Du
 * --------------
 * @notes: 输出到指定的socket地址
 **/
public class WriteToSocket {
    public static void main(String[] args) throws Exception {
        //local模式默认的并行度是当前机器的逻辑核的数量
        Configuration configuration = new Configuration();
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(configuration);
        int parallelism0 = env.getParallelism();
        System.out.println("执行环境默认的并行度：" + parallelism0);
        DataStreamSource<String> lines = env.socketTextStream("node1", 9999);
        //获取DataStream的并行度
        int parallelism = lines.getParallelism();
        System.out.println("SocketSource的并行度：" + parallelism);
        //lines.writeAsText("file:///Users/xing/Desktop/out");
        lines.writeToSocket("localhost", 9999, new SimpleStringSchema());
        env.execute();
    }
}
